<?php
/**
 * 流程控制函数
 * Created by PhpStorm.
 * User: hUsr
 * Date: 2018/11/9
 * Time: 10:49
 */

/**
 * 初始化，读取所有历史行情数据，计算指数，买卖点
 * @param $my_conn
 * @param $mongo
 */
function init($my_conn, $mongo)
{
    global $server, $stock, $exright; // 股票基本信息，行权记录
    $data = array(); // 所有股票的行情数据，二维数组，以股票代码为键，日期为子键
    $rt_temp = array(); // 实时行情临时数据，用于获取收盘价格
    echo "开始初始化……\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "开始初始化……"]);
    $c_time_start = getCurrentTime();

    // 删除mysql中旧的交易数据
    $sql = "DELETE FROM stock_transact";
    $my_conn->query($sql);

    // 删除临时数据集合
    foreach (['BUY_SELL', 'BUY_SELL2', 'MA', 'MA2', 'EMA', 'EMA2', 'MACD', 'MACD2', 'Last', 'Stock_quote'] as $item) {
        $collection = $mongo->stock->$item;
        $collection->drop();
    }

    // 标记指令状态
    $sql = "update stock_invoke set invoke_transact_state = 2";
    $my_conn->query($sql);
    // todo debug
//    $sql = "update stock_invoke set invoke_transact_state = 0 where Invoke_Type = 1 and Invoke_Date_Begin >= '2018-07-01'";
//    $my_conn->query($sql);

    // 获取最早的开始日期，最晚的结束日期
    $sql = 'select min(Invoke_Date_Begin) as Invoke_Date_Begin, max(Invoke_Date_End) as Invoke_Date_End, 
min(Invoke_Begin_Time) as Begin_Time, max(Invoke_Begin_Time) as End_Time, max(Stock_Invoke_Id) as Stock_Invoke_Id 
from stock_invoke where Invoke_Type = 1 and invoke_transact_state = 2';
    $res = $my_conn->query($sql);
    $res = mysqli_fetch_assoc($res);
    $begin_date = $res['Invoke_Date_Begin']; // 最早开始日期
    $end_date = $res['Invoke_Date_End']; // 最晚结束日期
    $begin_time = $res['Begin_Time']; // 分表开始时间
    $end_time = $res['End_Time']; // 分表结束时间
    $last_id = $res['Stock_Invoke_Id']; // 最后一个指令ID

    // 读取行情数据，所有股票，全部历史数据
    echo "开始读取行情数据……\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "开始读取行情数据……"]);
    $res = stock_quote($my_conn, $begin_date, $end_date, $begin_time, $end_time);
    echo "所有股票的全部数据读取完成，累计：" . (getCurrentTime() - $c_time_start) . "秒\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "所有股票的全部数据读取完成，累计：" . (getCurrentTime() - $c_time_start) . "秒"]);

    // 把实时行情转换成历史行情数据
    $c_time_start2 = getCurrentTime();
    echo "\n开始转换成历史行情……\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "开始转换成历史行情……"]);
    foreach ($res as $code => $row) {
        // 把实时行情转换成历史行情数据
        $generate = to_history($row);
        foreach ($generate as $value) {
            $data[$code][$value['date']] = $value; // 把每条记录压入以股票代码为键，日期为子键的二维数组
        }
        // 如果最后一个数据是实时行情，保存到$rt_temp
        $last = end($row);
        if ($last['Rt_Type'] == 2) {
            $rt_temp[$code] = $last;
        } else {
            unset($rt_temp[$code]);
        }
    }
    echo "转换成历史行情完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "转换成历史行情完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒"]);

    // 补充可能缺少的历史行情
    $c_time_start2 = getCurrentTime();
    echo "\n手动补充历史行情数据……\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "手动补充历史行情数据……"]);
    $dir = '/Zdata/re_stock/stock_quote/'; // 历史行情文本文件目录，要以“/”结尾
    foreach ($data as $code => $datum) {
        $result = read_history($dir, $code); // 读取补充的历史行情数据
        if ($result) {
            $data[$code] = array_merge($data[$code], $result); // 合并数据，结果数组的排序顺序可能不定按照日期先后
            // 按照日期排序
            $keys = array_keys($data[$code]); // 取得所有键
            $temp = array();
            $tmp = array();
            foreach ($keys as $val) {
                $temp[] = strtotime($val); // 转换成时间戳
            }
            sort($temp); // 从小到大排序时间戳
            foreach ($temp as $v) {
                $str = date('Y-m-d', $v); // 转换成日期格式
                $tmp[$str] = $data[$code][$str]; // 依次压入临时数组
            }
            $data[$code] = $tmp; // 替换原来未排序的
        }
    }
    echo "手动补充历史行情完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "手动补充历史行情完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒"]);

    // 循环所有股票，计算指数和买卖点
    $c_time_start2 = getCurrentTime();
    $num = count($data);
    echo "\n开始计算指数和买卖点，共计{$num}支股票\n";
    stat_broadcast("开始计算指数和买卖点，共计{$num}支股票");
    $i = 0;
    foreach ($data as $code => $value) {
        $i++;
        $exright[$code] = isset($exright[$code]) ? $exright[$code] : null; // 单个股票的行权记录
        process_stock($my_conn, $mongo, $stock[$code], $exright[$code], $code, $value, 1, true); // 初始化处理行情数据
        echo "{$code}计算完成，完成{$i}|总数{$num}，用时" . (getCurrentTime() - $c_time_start2) . "秒\n";
        $server->task(['type' => 'stat_broadcast', 'msg' => "{$code}计算完成，完成{$i}|总数{$num}，用时" . (getCurrentTime() - $c_time_start2) . "秒"]);
    }

    // 写入mongoDB
    $c_time_start2 = getCurrentTime();
    echo "\n准备写入mongoDB\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "准备写入mongoDB"]);
    // 写入历史行情，只保存了初始化过的历史行情数据，和已经转换成历史行情的实时行情数据
    $collection = $mongo->stock->Stock_quote;
    $collection->drop();
    $temp = array();
    foreach ($data as $key => $value) {
        $tmp = $value;
        $tmp['_id'] = $key;
        $temp[] = $tmp;
    }
    if ($temp) {
        $collection->insertMany($temp); // 保存新数据
    }
    echo "历史行情写入mongoDB完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "历史行情写入mongoDB完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒"]);

    // 写入最后的实时行情，最后的指令日期到Last
    $c_time_start2 = getCurrentTime();
    $collect = "Last";
    $collection = $mongo->stock->$collect;
    $collection->drop();
    $temp = array();
    $temp[] = ['_id' => 'invoke', 'date' => $end_date, 'invoke_id' => $last_id];
    foreach ($rt_temp as $key => $value) {
        $tmp = $value;
        $tmp['_id'] = $key;
        $temp[] = $tmp;
    }
    if ($temp) {
        $collection->insertMany($temp); // 保存新数据
    }
    echo "Last写入mongoDB完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "Last写入mongoDB完成，用时：" . (getCurrentTime() - $c_time_start2) . "秒"]);

    echo "\n初始化完成，累计用时：" . (getCurrentTime() - $c_time_start) . "秒\n\n";
    $server->task(['type' => 'stat_broadcast', 'msg' => "初始化完成，累计用时：" . (getCurrentTime() - $c_time_start) . "秒"]);
}


/**
 * 处理单个股票行情数据
 * @param $my_conn
 * @param $mongo
 * @param $stock 股票基本信息
 * @param $exright 行权记录
 * @param $code 股票代码
 * @param $data 行情数据包，历史行情可以为多条记录，实时行情只能是1条，否则会逻辑错误
 * @param $rt_type 实时类型，1历史行情，2实时行情
 * @param $is_init 是否初始化
 */
function process_stock($my_conn, $mongo, $stock, $exright, $code, $data, $rt_type = 1, $is_init = false)
{
    // 初始化时，只处理历史行情，实时行情已经提前转换成历史行情
    if ($is_init && $rt_type == 2) {
        return;
    }
    // 实时行情只能是1条，否则会逻辑错误
    if ($rt_type == 2 && count($data) != 1) {
        return;
    }

    // 如果有上一交易日的实时行情的收盘数据，当做历史行情，计算指数和买卖点
    if (!$is_init) { // 初始化时，这种情况不存在
        $collection = $mongo->stock->Last;
        $document = $collection->findOne(['_id' => $code]);
        if ($document && $document->date != $data[0]['date']) { // 此处假定实时行情每次只处理一条数据，否则逻辑会错误
            // 转换成数组
            $temp = array();
            foreach ($document as $key => $val) {
                $temp[$key] = $val;
            }
            $collection->deleteOne(['_id' => $code]); // 删除以防止重复循环
            process_stock($my_conn, $mongo, $stock, $exright, $code, [$temp]); // 用上一日的最后一个实时数据，计算历史行情买卖点
        }
    }

    // 计算价格的字段名
    $price_str = 'close'; // 历史行情用close字段

    // 实时行情，非初始化时才有，每次都替换保存的数据，每天产生买卖点后不再计算
    if ($rt_type == 2) {
        // 保存实时行情的最后一次数据，此处假定实时行情每次只处理一条数据，否则逻辑会错误
        $collection = $mongo->stock->Last;
        $tmp = $data[0]; // 保存未复权的数据
        $tmp['close'] = $tmp['price']; // 把实时价格保存为收盘价格
        $tmp['Rt_Type'] = 1; // 改为历史行情类型
        $tmp['_id'] = $code; // 以股票代码为主键
        $collection->deleteOne(['_id' => $code]); // 删除旧的数据
        $collection->insertOne($tmp); // 保存新数据
        // 跳过当天实时行情已经交易过的
        $collection = $mongo->stock->BUY_SELL2;
        $document = $collection->findOne(['_id' => $code]);
        if ($document) {
            $last_date = $document[0]->date;
            if (strtotime($last_date) == strtotime($data[0]['date'])) {
                return;
            }
        }
        // 实时行情取price字段
        $price_str = 'price';
        // 实时时段和间隔
        $rt_segment = $data[0]['Rt_Segment'];
        $rt_interval = $data[0]['Rt_Interval'];
    }


    ############################################# 开始计算
    // 复权计算
    $temp = array(); // 复权后的临时数据
    $origin = array(); // 原始数据
    foreach ($data as $item) {
        $origin[$item['date']] = $item; // 以日期为键保存原始数据
        if ($exright) {
            $item['close'] = exrightCalculate($item[$price_str], $item['date'], $exright); // 替换成复权后的价格
        } else {
            $item['close'] = $item[$price_str]; // 历史行情用close字段，实时行情用price字段
        }
        $temp[] = $item;
    }

    // 指数计算
    create_MA($mongo, $code, $temp, $rt_type);
    create_EMA($mongo, $code, $temp, $rt_type);
    create_DEA_MACD($mongo, $code, $rt_type);
    // 读取MA_MACD
    $ma_macd = read_MA_MACD($mongo, $code, $rt_type);

    // 读取已经保存的买卖点数据和交易状态，每次都以历史行情的计算结果为参考
    $collection = $mongo->stock->BUY_SELL;

    // 交易状态，买入后，保存买入价格，卖出后，为null
    if ($is_init) {
        $stat = null;
    } else {
        $document = $collection->findOne(['_id' => $code]);
        $stat = isset($document['stat']) ? $document['stat'] : null;
    }

    // 买卖点计算的组合条件
    $where = array(); // 买卖点计算的组合条件
    $where['macd'] = $stock['m']; // MACD1-20
    $where['x_macd'] = $stock['xm']; // MACD系数0-10
    $where['short'] = $stock['s']; // 短均线MA1-19
    $where['long'] = $stock['l']; // 长均线MA2-20
    $where['x_ma'] = $stock['x']; // 日均线系数1，0-10，买入条件
    $where['x_ma2'] = $stock['xx']; // 日均线系数2，0-10，买出条件

    // 计算买卖点
    $buy_sell = array(); // 买卖点数据
    foreach ($ma_macd as $value) {
        $result = calculate_buysell($where, $value, $stat); // 计算买卖点
        if ($result) {
            $tmp = $result['out']; // 计算结果
            $tmp['exright'] = $tmp['price']; // 复权后的价格
            $tmp['price'] = $origin[$tmp['date']]['close']; // 原始价格
            $buy_sell[] = $tmp;
            $stat = $result['stat'];
        }
    }

    // 保存数据
    if ($buy_sell) {
        $temp = array();
        // 已经保存的买卖点数据，只在非初始化，历史行情处理时，合并数据
        if (!$is_init && ($rt_type == 1) && $document) {
            // 转换对象为数组
            foreach ($document as $value) {
                // 转换对象为数组
                if (is_object($value)) {
                    $tmp = array();
                    foreach ($value as $key => $val) {
                        $tmp[$key] = $val;
                    }
                    $temp[] = $tmp;
                }
            }
        }
        // 合并数据
        foreach ($buy_sell as $item) {
            $temp[] = $item;
        }

        if ($rt_type == 2) {
            // 实时行情，保存到BUY_SELL2
            $collection = $mongo->stock->BUY_SELL2;
        }
        $temp['_id'] = $code; // 股票代码
        $temp['stat'] = $stat; // 交易状态
        $collection->deleteOne(['_id' => $code]); // 删除旧的数据
        $collection->insertMany([$temp]); // 合并后的买卖点写入mongoDB
        // 写入买卖点到mysql
        $buy_sell['_id'] = $code; // 股票代码
        $rt_segment = isset($rt_segment) ? $rt_segment : 'null'; // 实时时段
        $rt_interval = isset($rt_interval) ? $rt_interval : 'null'; // 实时间隔
        buysell_to_mysql($my_conn, $buy_sell, $rt_type, $rt_segment, $rt_interval); // 只写入本次计算结果
    }
}


/**
 * 检查流程控制表，读取新行情指令
 * @param $my_conn mysql连接实例
 * @param $mongo
 * @return bool
 */
function invoke($my_conn, $mongo)
{
    global $stock; // 基本信息
    $exright = get_exright($mongo); // 行权数据

    // 循环执行行情指令
    $sql = 'select Stock_Invoke_Id, Invoke_Date_Begin, Invoke_Date_End, Invoke_Begin_Time, Invoke_End_Time, 
Rt_Type, Rt_Segment, Rt_Interval from stock_invoke where Invoke_Type = 1 and invoke_transact_state = 0 order by Invoke_Date_Begin';
    $res = $my_conn->query($sql);
    while ($row = mysqli_fetch_assoc($res)) {
        $c_time_start = getCurrentTime();
        echo "开始处理行情指令{$row['Stock_Invoke_Id']}，实时类型{$row['Rt_Type']}，开始日期{$row['Invoke_Date_Begin']}，
        结束日期{$row['Invoke_Date_End']}，时段{$row['Rt_Segment']}，间隔{$row['Rt_Interval']}\n";
        stat_broadcast("开始处理行情指令{$row['Stock_Invoke_Id']}，实时类型{$row['Rt_Type']}，开始日期{$row['Invoke_Date_Begin']}，
        结束日期{$row['Invoke_Date_End']}，时段{$row['Rt_Segment']}，间隔{$row['Rt_Interval']}");

        // 上一个行情指令结束日期
        $invoke_date = null;
        $collect = "Last";
        $collection = $mongo->stock->$collect;
        $document = $collection->findOne(['_id' => 'invoke']);
        if ($document) {
            $invoke_date = $document['date'];
        }

        // 如果是实时行情，并且是每天的第一条指令，每天更新行权数据
        if ($row['Rt_Type'] == 2 && $invoke_date != $row['Invoke_Date_Begin']) {
            // 更新行权数据
            $exright = re_exright($my_conn, $mongo);
        }

        // 保存每次指令日期到Last集合
        $temp = ['_id' => 'invoke', 'date' => $row['Invoke_Date_End'], 'invoke_id' => $row['Stock_Invoke_Id']];
        $collection->deleteOne(['_id' => 'invoke']);
        $collection->insertMany([$temp]);

        // 读取行情数据
        $result = stock_quote($my_conn, $row['Invoke_Date_Begin'], $row['Invoke_Date_End'], $row['Invoke_Begin_Time'],
            $row['Invoke_Begin_Time'], $row['Rt_Type'], $row['Rt_Segment'], $row['Rt_Interval']);

        // 开始处理行情数据
        if ($result) {
            echo "开始处理行情数据……\n";
            stat_broadcast("开始处理行情数据……");

            // 标记流程控制指令为处理中
            $sql = "update stock_invoke set invoke_transact_state = 1 where Stock_Invoke_Id = {$row['Stock_Invoke_Id']}";
            $my_conn->query($sql);

            // 处理所有股票行情数据
            foreach ($result as $code => $value) {
                // 处理单个股票数据
                $exright[$code] = isset($exright[$code]) ? $exright[$code] : null; // 单个股票的行权记录
                process_stock($my_conn, $mongo, $stock[$code], $exright[$code], $code, $value, $row['Rt_Type']); // 计算实时行情买卖点
            }

            echo "行情指令{$row['Stock_Invoke_Id']}完成，用时：" . (getCurrentTime() - $c_time_start) . "\n\n";
            stat_broadcast("行情指令{$row['Stock_Invoke_Id']}完成，用时：" . (getCurrentTime() - $c_time_start));

        } else {
            echo "没有行情数据\n";
            stat_broadcast("没有行情数据");
        }

        // 标记流程控制指令为处理完成
        $sql = "update stock_invoke set invoke_transact_state = 2 where Stock_Invoke_Id = {$row['Stock_Invoke_Id']}";
        $my_conn->query($sql);
    }
    if (!$row) {
        echo "没有行情消息\n";
        stat_broadcast("没有行情消息");
    }
}


